//package com.project.stream_consumer.demo10.controller;
//
//import com.project.stream_consumer.demo7.conf.MyInPut;
//import com.rabbitmq.client.Channel;
//import org.springframework.amqp.support.AmqpHeaders;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.cloud.stream.annotation.EnableBinding;
//import org.springframework.cloud.stream.annotation.StreamListener;
//import org.springframework.kafka.support.Acknowledgment;
//import org.springframework.kafka.support.KafkaHeaders;
//import org.springframework.messaging.Message;
//import org.springframework.messaging.handler.annotation.Header;
//import org.springframework.stereotype.Component;
//
//import java.io.IOException;
//import java.util.concurrent.TimeUnit;
//
//
//@Component
//@EnableBinding(MyInPut.class)
//public class ConsunmerController {
//    @Value("${spring.application.name}")
//    String name;
//    @StreamListener(MyInPut.MYINPUT1)
//    public void input1(Message<String> message,@Header(AmqpHeaders.CHANNEL) Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) Long deliveryTag){
//        System.out.println("input1:"+message.getPayload());
//        try {
//            channel.basicAck(deliveryTag, false);
//        } catch (IOException e) {
//            e.printStackTrace();
//        }
//    }
//    @StreamListener(MyInPut.MYINPUT2)
//    public void input2(Message<String> message){
//        System.out.println("input2:"+message.getPayload());
//    }
//    @StreamListener(MyInPut.MYKAFKAINPUT)
//    public void input3(Message<String> message) throws InterruptedException {
//        Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
//        if (acknowledgment != null) {
//            TimeUnit.SECONDS.sleep(1);
//            System.out.println(Thread.currentThread().getName()+"--->"+name+">kafka:"+message.getPayload());//接收到的消息
//            acknowledgment.acknowledge();//确认收到消息
//        }
//    }
//}